package com.toutiao.issue

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.{HashMap => HMap}
import scala.collection.mutable.{HashSet => HSet}

import java.util.Date

import org.apache.log4j.Logger

import com.toutiao.AppProp

class IssueModel(
		source: EventStore,
		es: EventStore,
		map: MapStore,
		topU: Double,
		N: Int, //时间窗口大小
		locationDate: String // 记录时间文件位置
    ) {
  
  private
  val writeMap = 0
  
  private
  val _log = Logger.getLogger(this.getClass.getName)
  
  private
  val M = new HMap[String,String]  //EventID 
  
  private
  val M1 = new HMap[String,Int]
  private
  val M2 = new HMap[Int,String]
  private
  var N1 = 0
  private
  var N2 = 0
  
  val Data = source
  
  private 
  val Ts = new TimeArray
  
  def maxTime = Ts.maxTime
  def minTime = Ts.minTime
  
  val IssueData = new IssueCollector(topU)
  
  _log.info("WAND S "+topU)
  val ES = es

  def processNewOne(e:Event,minCnt:Int) = synchronized {
    if (!M1.contains(e.name)){ // 首次
      N2 += 1
      M1(e.name) = N2
      M2(N2) = e.name
      val ns = IssueData.processNewOne(M1,M2,e,topU,minCnt)
      val name = 
        if (ns.size > 0) ns.last
        else ""
      if (ns.size > 1){
        _log.info("Merge "+ns)
      }
      if (name != ""){
        Ts.insert(e.name, e.T1)
        //Tm(e.name) = e.T1
        M(e.name) = name
        IssueData(name).ms.map(_.split(" ")(0)).filter(M.contains(_)).foreach(n => {
          M(n) = name
        })
//        ns.foreach(n => {
//          M(n) = name
//        })
      }
      else {
        M1 -= e.name
        M2 -= N2
        N2 -= 1
      } 
      //name
      ns
    } 
    else { // 已经存在需要更新
      val name = M(e.name)
      IssueData.updateMs(name, e)
      List(name)
    }
  }
  
  def syncOne(c:String) = {
    val us = IssueData(c).ms.map(_.split(" ")(0))
    us.map(u => {
      (u,c,IssueData(c).ct,IssueData(c).ut)
    })
  }

  def process(ms:Array[Event],minCnt:Int):Array[String] = {
    val ns = ms.flatMap(e => processNewOne(e,minCnt))
    ns
  }
  
  def saveOne(n:String) {
    if (IssueData.isIn(n)){
      val us = IssueData(n).ms.map(x => {
        val u = x.split(" ")(0)
        u
      }).toSet.toList
      ES.insert(IssueData(n),us)
    }
  }
  
  def save(ns:Array[String]) {
    ns.foreach(n => saveOne(n))
  }
  
  def remove(T:Date) = synchronized {
    val i = Ts.before(T)
    if (i < 1){
      0
    }
    else {
      //val ms = Ts.start(i).filter(m => M1.contains(m._1))
      val ms = Ts.start(i)
      _log.info("remove "+ms.size)
      for (m <- ms if M1.contains(m._1)){
        val name = m._1
        val index = M1(name)
        if (IssueData.isIn(name)){
          IssueData.remove(name,index)
        }
        M1 -= name
        M2 -= index
        M -= name
        //if (Tm.contains(name))
        //  Tm -= name
      }
      N1 += ms.size
      Ts.remove(i)
      ms.size
    }
  }
  
  // !!! 这里有问题
  private
  def _init(xs:Array[Event]) = synchronized {
    for (x <- xs if x.ms.size > 0){
      val us = x.ms.map(_.split(" ")(0)).toSet+x.name
      for (u <- us){
    	  N2 += 1
    	  M1(u) = N2
          M2(N2) = u
          M(u) = x.name
          Ts.insert(u, x.T1)
          //Tm(u) = x.T1
      }
      IssueData.insertData(x.name,x)
      IssueData.insertIndex(M1(x.name),x.vs.map(_._1).toArray)
    }
  }
  
  def init {
    val current = SimpleDate.loadDate(locationDate)
    val T1 = SimpleDate.increment(current, -N*60)
    val T2 = SimpleDate.increment(current, 60)
    _log.info(SimpleDate.format(T1)+" => "+SimpleDate.format(T2))
    val xs = ES.retrieve(T1,T2,"UPDATE_TIME","MS")
    
    _init(xs)
    
    _log.info("MAX_ID "+N2)
    _log.info("MIN_ID "+N1)
  }
  
  // 批处理运行
  def run(m:Int,minCnt:Int) = synchronized {
    var current = SimpleDate.loadDate(locationDate)
    var eT = Data.getMaxTime("UPDATE_TIME")
    if (eT.getTime > SimpleDate.current.getTime)
      eT = SimpleDate.current
    
    while (1 == 1) {
    //while (current.getTime < SimpleDate.current.getTime) {
      _log.info(SimpleDate.format(eT))
      //val S = SimpleDate.current
      val x = SimpleDate.increment(current,m)
      if (x.getTime <= eT.getTime) {
        _log.info("Current Date "+SimpleDate.format(current))
        _log.info("End Date "+SimpleDate.format(eT))
        _log.info(SimpleDate.format(Ts.maxTime))
        _log.info(SimpleDate.format(Ts.minTime))
        val T1 = current
        val T2 = SimpleDate.increment(T1,m)
        
        val ms = Data.retrieve(T1,T2,"UPDATE_TIME","US")
        val ns = process(ms,minCnt)
        
        _log.info("New Event "+ns.size)
        
        _log.info("Save ... "+ns.size)
        if (ns.size > 0)
        	save(ns)
        _log.info("Save Done")
        
        
        val rs = ns.filter(IssueData.isIn(_)).flatMap(syncOne(_))
        _log.info("Save Map ... "+rs.size)
        if (rs.size > 0 && writeMap > 0)
        	map.insert(rs)
        _log.info("Save Map Done")
        
        val T = SimpleDate.increment(current,-60*N)
        _log.info(SimpleDate.format(T))
        val n = remove(T)
        _log.info("Remove News "+n)
        _log.info("Current Size "+IssueData.size)
        _log.info("Current Event Size "+M.size)
        _log.info("MAX_ID "+N2)
        _log.info("MIN_ID "+N1)
        val now = SimpleDate.current
        current = 
          if (Ts.maxTime.getTime > current.getTime) 
            if (Ts.maxTime.getTime <= now.getTime)
              Ts.maxTime
            else
              now
          else T2
        SimpleDate.saveDate(current,locationDate)
      }
      else {
        val T = SimpleDate.current
        _log.info("Sleep Here "+SimpleDate.format(T))
        Thread.sleep(30*1000)
        eT = Data.getMaxTime("UPDATE_TIME")
        if (eT.getTime > SimpleDate.current.getTime)
        	eT = SimpleDate.current        
      }
    }    
  }
}


object IssueProcess {
  
//  def run(source: EventStore,
//		  esName:String,esUri:String,esPort:Int,esIndexName:String,esTypeName:String,
//		  esMapName:String,esMapUri:String,esMapPort:Int,esMapIndex:String,esMapType:String,
//		  N:Int, locationDate:String,U:Double, K:Int,C:Int)
  def run(prop:AppProp){
    val esSourceName = prop("ES_NAME")
    val esSourceURI = prop("ES_URI")
    val esSourcePort = prop("ES_PORT").toInt
    val esSourceIndex = prop("ES_INDEX")
    val esSourceType = prop("ES_TYPE")
    val source = new EventStore(esSourceName, esSourceURI, esSourcePort, esSourceIndex, esSourceType)
    
    val esName = prop("ES_CONTEXT_NAME")
    val esUri = prop("ES_CONTEXT_URI")
    val esPort = prop("ES_CONTEXT_PORT").toInt
    val esIndexName = prop("ES_CONTEXT_INDEX")
    val esTypeName = prop("ES_CONTEXT_TYPE")
    
    val esMapName = prop("ES_MAP_NAME")
    val esMapUri = prop("ES_MAP_URI")
    val esMapPort = prop("ES_MAP_PORT").toInt
    val esMapIndex = prop("ES_MAP_INDEX")
    val esMapType = prop("ES_MAP_TYPE")
    val es = new EventStore(esName,esUri,esPort,esIndexName,esTypeName)
    val map = new MapStore(esMapName,esMapUri,esMapPort,esMapIndex,esMapType)
    
     val locationDate = prop("TIME_LOCATION")
     
     val N = prop("TIME_SIZE").toInt
     val U = prop("TOP_S").toDouble
     val K = prop("TIME_STEP").toInt
     val C = prop("MIN_COUNT").toInt
  
     val m = new IssueModel(source,es,map,U,N,locationDate)
    
    	m.init
    	m.run(K,C)
  }
  
}
